package com.smdp.service.impl;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.oss.ClientException;
import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.exceptions.ServerException;
import com.aliyuncs.green.model.v20180509.TextScanRequest;
import com.aliyuncs.http.FormatType;
import com.aliyuncs.http.HttpResponse;
import com.aliyuncs.profile.DefaultProfile;
import com.aliyuncs.profile.IClientProfile;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.smdp.utils.BlogData;
import com.smdp.entity.Blog;
import com.smdp.service.IBlogService;
import com.smdp.service.VerifyText;
import com.smdp.utils.SimpleRedisLock;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;

import static com.smdp.utils.AliyunConstants.ALIYUN_OPEN_APP_ID;
import static com.smdp.utils.AliyunConstants.ALIYUN_OPEN_APP_SECRET;
import static com.smdp.utils.BlogConstants.*;
import static com.smdp.utils.RabbitMQConstants.VERIFY_QUEUE;

/**
 * 审核业务实现类。
 * 包括监听RabbitMQ的审核队列、文本审核功能（通过阿里云服务实现，开启独立线程执行）+审核审核通过则发布/更新博客。
 *
 * @author 蔡方珩
 */
@Slf4j
@Service
public class VerifyTextImpl implements VerifyText {

    @Resource
    IBlogService blogService;

    @Resource
    StringRedisTemplate stringRedisTemplate;

    /**
     * 处理审核业务的线程池
     */
    private static final ThreadPoolExecutor BLOG_VERIFY_EXECUTOR = new ThreadPoolExecutor(
            2, //核心线程池大小
            8, //最大线程数量，获取CPU核数 System.out.println(Runtime.getRuntime().availableProcessors());
            3, //超时时间，没人调用时就会释放
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(40),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy()
    );

    /**
     * 监听MQ，开启新线程审核blog信息，通过审核的则发布博客
     * @param message 消息
     * @param channel channel
     */
    @Override
    @RabbitListener(queues = VERIFY_QUEUE)
    public void handleUnverifiedBlog(Message message, Channel channel) throws IOException {
        // 1.1.接收消息
        ObjectMapper mapper=new ObjectMapper();
        String json = new String(message.getBody());
        BlogData blogData =mapper.readValue(json.getBytes(StandardCharsets.UTF_8), BlogData.class);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        Blog blog = blogData.getBlog();
        // 2.1.获取互斥锁
        String lockKey = BLOG_VERIFY_LOCK_KEY + blog.getId();
        SimpleRedisLock lock = new SimpleRedisLock(lockKey, stringRedisTemplate);
        boolean isLock = lock.tryLock(20);
        // 2.2.判断是否获取锁成功
        if (isLock) {
            //2.3.成功，开启独立线程，处理博客审核
            BLOG_VERIFY_EXECUTOR.submit(() -> {
                try {
                    // 博客审核
                    boolean pass = this.verify(blog.getContent());
                    if (!pass){
                        return;
                    }
                    // 通过审核，保存/更新博客
                    if (Objects.equals(blogData.getStatus(), NEED_SAVE)){
                        blogService.saveBlogAfterVerify(blog);
                    }
                    if (Objects.equals(blogData.getStatus(), NEED_UPDATE)){
                        blogService.updateBlogAfterVerify(blog);
                    }
                } catch (Exception e) {
                    throw new RuntimeException(e);
                } finally {
                    // 释放锁
                    lock.unlock();
                }
            });
        }
    }

    /**
     * 通过阿里云反垃圾检测审核文本
     * @param text 文本
     * @return 通过：true / 未通过或出错：false
     */
    public boolean verify(String text) throws Exception {
        IClientProfile profile = DefaultProfile
                .getProfile("cn-shanghai", ALIYUN_OPEN_APP_ID, ALIYUN_OPEN_APP_SECRET);
        DefaultProfile
                .addEndpoint("cn-shanghai", "Green", "green.cn-shanghai.aliyuncs.com");
        IAcsClient client = new DefaultAcsClient(profile);
        TextScanRequest textScanRequest = new TextScanRequest();
        textScanRequest.setAcceptFormat(FormatType.JSON); // 指定API返回格式。
        textScanRequest.setHttpContentType(FormatType.JSON);
        textScanRequest.setMethod(com.aliyuncs.http.MethodType.POST); // 指定请求方法。
        textScanRequest.setEncoding("UTF-8");
        textScanRequest.setRegionId("cn-shanghai");
        List<Map<String, Object>> tasks = new ArrayList<Map<String, Object>>();
        Map<String, Object> task1 = new LinkedHashMap<String, Object>();
        task1.put("dataId", UUID.randomUUID().toString());
        /*
         * 待检测的文本，长度不超过10000个字符。
         */
        task1.put("content", text);
        tasks.add(task1);
        JSONObject data = new JSONObject();

        /*
         * 检测场景。文本垃圾检测则传递antispam。
         */
        data.put("scenes", Arrays.asList("antispam"));
        data.put("tasks", tasks);
        JSON.toJSONString(data, true);
        textScanRequest.setHttpContent(data.toJSONString().getBytes(StandardCharsets.UTF_8), "UTF-8", FormatType.JSON);
        // 请务必设置超时时间。
        textScanRequest.setConnectTimeout(3000);
        textScanRequest.setReadTimeout(6000);
        boolean pass = false;
        try {
            HttpResponse httpResponse = client.doAction(textScanRequest);
            if (httpResponse.isSuccess()) {
                // 如果成功响应，继续
                JSONObject scrResponse = JSON.parseObject(new String(httpResponse.getHttpContent(), StandardCharsets.UTF_8));
                if (200 == scrResponse.getInteger("code")) {
                    // 如果状态码为200，继续
                    JSONArray taskResults = scrResponse.getJSONArray("data");
                    for (Object taskResult : taskResults) {
                        if (200 == ((JSONObject) taskResult).getInteger("code")) {
                            JSONArray sceneResults = ((JSONObject) taskResult).getJSONArray("results");
                            for (Object sceneResult : sceneResults) {
                                // 获取判定结果
                                String suggestion = ((JSONObject) sceneResult).getString("suggestion");
                                if (suggestion.equals("pass")){
                                    pass = true;
                                }
                            }
                        } else {
                            log.error("task process fail:" + ((JSONObject) taskResult).getInteger("code"));
                        }
                    }
                    return pass;
                } else {
                    log.error("detect not success. code:" + scrResponse.getInteger("code"));
                }
            } else {
                log.error("response not success. status:" + httpResponse.getStatus());
            }
        } catch (ServerException e) {
            e.printStackTrace();
        } catch (ClientException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return pass;
    }

}
